package com.nit.netty.zookeeper.client;

import com.nit.netty.manager.NettyChannelManager;
import com.nit.netty.worker.NettyThreadPool;
import lombok.extern.slf4j.Slf4j;
import net.jodah.expiringmap.ExpirationPolicy;
import net.jodah.expiringmap.ExpiringMap;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.*;
import org.apache.zookeeper.Watcher.Event.*;

import javax.servlet.ServletContextListener;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: whb
 * @Description:
 * @CreateTime: 2020/8/24 17:13
 */
@Slf4j
public class ZookeeperClient implements Watcher, ServletContextListener {
    private ZooKeeper zk;
    private CuratorFramework zkClient;
    private static final int SESSION_TIMEOUT = 15000;
    // 信号量,zk连接异步，用于阻塞，保证所有的操作都在zk创建连接成功以后再执行
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);
    private NettyChannelManager manager;
    private static final String REGISTRY_PATH = "/registry";
    private String zkAddress;
    // 初始值为null, 加载完毕zookeeper后给赋值
    private ExpiringMap<String, String> expiringMap = null;

    public ZookeeperClient() {

    }

    public ZookeeperClient(String zkAddress) {
        connect(zkAddress);
    }

    /**
    * @Description: 由于会话断开后需要重连, 所以从构造器中提取出了connect
    */
    private void connect(String zkAddress) {
        try {
            expiringMap = null;
            // 监听器为自己当前类
            zk = new ZooKeeper(zkAddress, SESSION_TIMEOUT, this);
            this.zkAddress = zkAddress;
            countDownLatch.await();
            // 在第一次注册监听器的时候, map为空, 所以不会触发refresh
            // 初始化一个带有时限的map
            // 每一个服务器ip为key, value可以不要
            // 延迟1秒可以保证该服务器的所有服务注册完毕, 这样可以防止重复refresh
            // 两秒后map中的entry销毁, 监听器监听到销毁事件来进行refresh
            expiringMap = ExpiringMap.builder()
                    .maxSize(100)
                    .expiration(1L, TimeUnit.SECONDS)
                    .expirationPolicy(ExpirationPolicy.ACCESSED)
                    .variableExpiration()
                    .expirationListener((key, value) -> {
                        log.info("ip:{}下所有服务已经连接", key);
                        manager.refresh();
                    })
                    .build();
            // 断开重连的refresh
            if(null != manager) {
                manager.refresh();
            }
        } catch (Exception ex) {
            log.error("zookeeper连接失败", ex);
        }
    }

    /**
    * @Description: 由于ExpiringMap没有public的构造器, 所以无法实现内部类, 只能默认用个put方法
    */
    private void addToMap(String key) {
        String DEFAULT_VALUE = "DEFAULT_VALUE";
        if(null != expiringMap) {
            expiringMap.put(key, DEFAULT_VALUE);
        }
    }

    public void setManager(NettyChannelManager manager) {
        this.manager = manager;
    }

    /**
     * 获取路径下所有子节点
     *
     * @param
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    public List<String> getAllService() throws KeeperException, InterruptedException {
        List<String> children = zk.getChildren(REGISTRY_PATH, false);
        return children;
    }

    public List<String> getServiceByName(String serviceName) throws KeeperException, InterruptedException {
        String path = REGISTRY_PATH + "/" + serviceName;
        List<String> children = zk.getChildren(path, false);
        return children;
    }

    public String getData(String nodeName) throws KeeperException, InterruptedException {
        String path = REGISTRY_PATH + "/" + nodeName;
        byte[] data = zk.getData(path, false, null);
        if (data == null) {
            return "";
        }
        return new String(data);
    }

    private void refreshManager(String ip) {
        // map为空则表示这是第一次注册监听器
        if (null != expiringMap) {
            // 重复put可以刷新过期时间, entry过期后refresh
            addToMap(ip);
        }
    }

    private void addRootListener(String root, CuratorFramework zkClient) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(zkClient, root, true);
        // 在初始化时就开始进行监听
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        // 我们所希望的是监听中的代码块执行完毕后, 才执行我们的业务逻辑, 所以加入线程池监听状态
        cache.getListenable().addListener((client, event) ->
            NettyThreadPool.getInstance().submit(() -> {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        String serverPath = event.getData().getPath();
                        log.info("新增节点：" + serverPath);
                        //监听Server, 在具体服务获取时做refresh
                        try {
                            addServerListener(serverPath, client);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        break;
                    case CHILD_UPDATED:
                        log.info("节点：" + event.getData().getPath() + ",数据修改为：" + new String(event.getData().getData()));
                        break;
                    case CHILD_REMOVED:
                        String serverPath2 = event.getData().getPath();
                        log.info("节点：" + serverPath2 + "被删除");
                        break;
                    default:
                        break;
                }
            }));
    }


    private void addServerListener(String path, CuratorFramework zkClient) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(zkClient, path, true);
        // 在初始化时就开始进行监听
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener((client, event) -> NettyThreadPool.getInstance().submit(() -> {
            switch (event.getType()) {
                case CHILD_ADDED:
                    byte[] childData = event.getData().getData();
                    String ip = new String(childData);
                    log.info("新增子节点：" + event.getData().getPath() + ",数据为：" + ip);
                    // 调用一个refresh方法, 判断逻辑在方法中
                    refreshManager(ip);
                    break;
                case CHILD_UPDATED:
                    log.info("子节点：" + event.getData().getPath() + ",数据修改为：" + new String(event.getData().getData()));
                    break;
                case CHILD_REMOVED:
                    byte[] childData2 = event.getData().getData();
                    String ip2 = new String(childData2);
                    log.info("子节点：" + event.getData().getPath() + "被删除");
                    refreshManager(ip2);
                    break;
                default:
                    break;
            }
        }));
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        // 获取事件状态
        KeeperState keeperState = watchedEvent.getState();
        // 获取事件类型
        EventType eventType = watchedEvent.getType();
        // 判断是否建立连接
        if (KeeperState.SyncConnected == keeperState) {
            // 如果当前状态已经连接上了 SyncConnected：连接，AuthFailed：认证失败,Expired:失效过期,
            // ConnectedReadOnly:连接只读,Disconnected:连接失败
            if (EventType.None == eventType) {
                // 开启一个监听服务
                try {
                    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
                    this.zkClient = CuratorFrameworkFactory.builder()
                            .connectString(zkAddress)
                            .sessionTimeoutMs(SESSION_TIMEOUT)
                            .retryPolicy(retryPolicy).build();
                    this.zkClient.start();
                    addRootListener(REGISTRY_PATH, zkClient);
                    // 保证线程池内添加了任务
                    Thread.sleep(500);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 通过线程池的活动任务数判断所有的监听器已经执行完毕
                    // 即此刻节点扫描已经完成
                    for(;;) {
                        int activeCount = NettyThreadPool.getInstance().getActiveCount();
                        if(activeCount == 0) {
                            break;
                        }
                    }
                    countDownLatch.countDown();
                }
                log.info("zookeeper连接成功, address:{}", zkAddress);
            }
        } else if (KeeperState.Expired == keeperState) {
            connect(zkAddress);
        }
    }
}
